package com.gaga.console.base.utils;

import com.gaga.common.enums.PushMessageType;
import com.gaga.console.base.entity.sse.PushMessageDto;
import com.gaga.metadata.common.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 * @author liujia
 */
@Slf4j
@Component
public class SseEmitterServerUtilsbak {

    private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
    /**
     * 心跳消息内容
     */
    private static final PushMessageDto HEARTBEAT_MESSAGE = new PushMessageDto(PushMessageType.HEARTBEAT, null, null, null, null, 0);

    /**
     * 心跳间隔时间，单位毫秒（例如30秒）
     */
    private static final long HEARTBEAT_INTERVAL = 30_000L;

    public static SseEmitter connect(String channelId) {
        //设置超时时间，0表示不过期，默认是30秒，超过时间未完成会抛出异常
        // 这个地方30分钟
        SseEmitter sseemitter = new SseEmitter(0L);
        //注册回调
        sseemitter.onCompletion(completionCallBack(channelId));
        sseemitter.onError(errorCallBack(channelId));
        sseemitter.onTimeout(timeoutCallBack(channelId));
        sseEmitterMap.put(channelId, sseemitter);
        log.info("create new sse connect ,current channelId:{}", channelId);
        return sseemitter;
    }

    public static void close(String channelId) {
        SseEmitter sseEmitter = getId(channelId);
        if (sseEmitter != null) {
            sseEmitter.complete();
            remove(channelId);
        }
    }


    public static void batchSendMessage(Object message) {
        sseEmitterMap.forEach((channelId, sseEmitter) -> {
            try {
                sseEmitter.send(message);
            } catch (IOException e) {
                log.error("channelId:{}, send message error:{}", channelId, e.getMessage());
                remove(channelId);
            }
        });
    }

    /**
     * 给指定用户发消息
     */
    public static void sendMessage(String channelId, Object message) {
        if (StringUtils.isBlank(channelId)) {
            return;
        }
        if (sseEmitterMap.containsKey(channelId)) {
            try {
                sseEmitterMap.get(channelId).send(message);
            } catch (Exception e) {
                sseEmitterMap.remove(channelId);
                log.error("channelId:{}, send message error:{}", channelId, e.getMessage());
                e.printStackTrace();
            }
        }
    }


    //移除用户
    public static SseEmitter remove(String channelId) {
        if (sseEmitterMap.containsKey(channelId)) {
            SseEmitter remove = sseEmitterMap.remove(channelId);
            log.info("remove channelId:{}", channelId);
            return remove;
        }
        return null;
    }

    /**
     * 定时发送心跳消息
     */
    @Scheduled(fixedRate = HEARTBEAT_INTERVAL)
    public void sendHeartbeat() {
        log.debug("Sending heartbeat to all connected clients(向所有连接的客户端发送心跳)");
        batchSendMessage(HEARTBEAT_MESSAGE);
    }


    public static SseEmitter getId(String channelId) {
        return sseEmitterMap.get(channelId);
    }

    private static Runnable completionCallBack(String channelId) {
        return () -> {
            log.info("结束连接,{}", channelId);
            remove(channelId);
        };
    }

    private static Runnable timeoutCallBack(String channelId) {
        return () -> {
            log.info("连接超时,{}", channelId);
            remove(channelId);
        };
    }

    private static Consumer<Throwable> errorCallBack(String channelId) {
        return (throwable) -> {
            log.error("连接异常，{}", channelId);
            remove(channelId);
        };
    }
}

